{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# `Reaqtor.IoT`\n",
        "\n",
        "Notebook equivalent of the Playground console application."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Reference the library\n",
        "\n",
        "We'll just import the entire console application to get the transitive closure of referenced assemblies."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "#r \"bin/Debug/net6.0/Reaqtor.IoT.dll\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## (Optional) Attach a debugger\n",
        "\n",
        "If you'd like to step through the source code of the library while running samples, run the following cell, and follow instructions to start a debugger (e.g. Visual Studio). Navigate to the source code of the library to set breakpoints."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "System.Diagnostics.Debugger.Launch();"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Import some namespaces"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "using System;\n",
        "using System.Linq;\n",
        "using System.Linq.CompilerServices.TypeSystem;\n",
        "using System.Linq.Expressions;\n",
        "using System.Threading;\n",
        "using System.Threading.Tasks;\n",
        "\n",
        "using Nuqleon.DataModel;\n",
        "\n",
        "using Reaqtive;\n",
        "using Reaqtive.Scheduler;\n",
        "\n",
        "using Reaqtor;\n",
        "using Reaqtor.IoT;"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Configure environment\n",
        "\n",
        "**Query engines** host reactive artifacts, e.g. subscriptions, which can be stateful.\n",
        "\n",
        "Query engines are a failover unit. State for all artifacts is persisted via checkpointing.\n",
        "\n",
        "Query engines depend on services from the environment:\n",
        "\n",
        "* A **scheduler** to process events on:\n",
        "  * There's one physical scheduler per host. Think of it as a thread pool.\n",
        "  * Each engine has a logical scheduler. Think of it as a collection of tasks. The engine suspends/resumes all work for checkpoint/recovery.\n",
        "* A **key/value store** for state persistence, including:\n",
        "  * A transaction log of create/delete operations for reactive artifacts.\n",
        "  * Periodic checkpoint state, which includes:\n",
        "    * State of reactive artifacts (e.g. sum and count for an Average operator).\n",
        "    * Watermarks for ingress streams, enabling replay of events upon failover.\n",
        "\n",
        "This sample also parameterizes query engines on an ingress/egress manager to receive/send events across the engine/environment boundary."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "To run query engines in the notebook, we write a simple `WithEngine` helper. This part takes care of setting up the engine and creating the environment. The general lifecycle of an engine is as follows.\n",
        "\n",
        "* Instantiate the object, passing the environment services.\n",
        "* Recover the engine's state from the key/value store.\n",
        "* Use the engine (through the `action` callback in the helper).\n",
        "* Checkpoint the engine's state. This is typically done periodically, e.g. once per minute. The interval is a tradeoff between:\n",
        "  * I/O frequency versus I/O size, e.g. due to state growth as events get processed.\n",
        "  * Replay capacity for ingress events and duration of replay, e.g. having to replay up to 1 minute worth of events from a source.\n",
        "* Unloading the engine. This is optional but useful for graceful shutdown. In the Reactor service this is used when a primary moves to another node in the cluster. It allows reactive artifacts to unload resources (e.g. connections)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "var store = new InMemoryKeyValueStore();\n",
        "var iemgr = new IngressEgressManager();\n",
        "\n",
        "async Task WithEngine(Func<MiniQueryEngine, Task> action)\n",
        "{\n",
        "    using var ps = PhysicalScheduler.Create();\n",
        "    using var scheduler = new LogicalScheduler(ps);\n",
        "    using var engine = new MiniQueryEngine(new Uri(\"iot://reactor/1\"), scheduler, store, iemgr);\n",
        "\n",
        "    using (var reader = store.GetReader())\n",
        "        await engine.RecoverAsync(reader);\n",
        "\n",
        "    await action(engine);\n",
        "\n",
        "    using (var writer = store.GetWriter())\n",
        "        await engine.CheckpointAsync(writer);\n",
        "\n",
        "    await engine.UnloadAsync();\n",
        "}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create a first empty engine\n",
        "\n",
        "Here we just load and use the engine for the first time. We don't do anything with it yet, but this will have the side-effect of initializing the key/value store used for checkpointing."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    // Do nothing.\n",
        "    await Task.Yield();\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Let's analyze the checkpoint store. We'll find a few engine-created artifacts in a couple of tables. The details aren't very important to us right now."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "Table 'TxMetadata':\r\n\r\n  Key 'Latest':\r\n    Value = 01 31\r\n    Size  = 14\r\n\r\n  Key 'ActiveCount':\r\n    Value = 01 31\r\n    Size  = 24\r\n\r\n  Key 'HeldCount':\r\n    Value = 01 31\r\n    Size  = 20\r\n\r\n\r\n  Size  = 78\r\n\r\nTable 'Observers':\r\n\r\n  Key 'mgmt://canary/observer':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 88 01 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 4D 60 22 2C 30 2C 22 4E 6F 70 22 2C 31 2C 5B 5D 2C 32 5D 2C 5B 22 4D 3C 3E 22 2C 30 2C 5B 33 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 4F 62 73 65 72 76 65 72 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 4F 62 73 65 72 76 65 72 60 31 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 2D 31 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 31 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 52 65 61 71 74 69 76 65 2E 43 6F 72 65 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 2E 28 29 22 2C 31 2C 5B 5D 5D 7D 18 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 65 72 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 649\r\n\r\n\r\n  Size  = 667\r\n\r\nTable 'Observables':\r\n\r\n  Key 'mgmt://canary/observable':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 E1 01 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 4D 60 22 2C 30 2C 22 45 6D 70 74 79 22 2C 31 2C 5B 5D 2C 32 5D 2C 5B 22 4D 3C 3E 22 2C 30 2C 5B 33 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 53 75 62 73 63 72 69 62 61 62 6C 65 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 62 61 62 6C 65 60 31 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 2D 31 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 32 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 52 65 61 71 74 69 76 65 2E 4C 69 6E 71 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 52 65 61 71 74 69 76 65 2E 49 6E 74 65 72 66 61 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 2E 28 29 22 2C 31 2C 5B 5D 5D 7D 1A 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 61 62 6C 65 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 744\r\n\r\n\r\n  Size  = 766\r\n\r\nTable 'Subscriptions':\r\n\r\n  Key 'mgmt://canary/subscription':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 39 02 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 46 75 6E 63 60 33 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 62 61 62 6C 65 60 31 22 2C 31 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 32 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 4F 62 73 65 72 76 65 72 60 31 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 34 2C 5B 32 5D 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 70 74 69 6F 6E 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 30 2C 5B 33 2C 35 2C 36 5D 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 2C 22 52 65 61 71 74 69 76 65 2E 49 6E 74 65 72 66 61 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 2C 22 47 6C 6F 62 61 6C 73 22 3A 5B 5B 37 2C 22 72 78 3A 2F 2F 6F 62 73 65 72 76 61 62 6C 65 2F 73 75 62 73 63 72 69 62 65 22 5D 2C 5B 33 2C 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 61 62 6C 65 22 5D 2C 5B 35 2C 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 65 72 22 5D 5D 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 28 29 22 2C 5B 22 24 22 2C 30 5D 2C 5B 5B 22 24 22 2C 31 5D 2C 5B 22 24 22 2C 32 5D 5D 5D 7D 1C 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 73 75 62 73 63 72 69 70 74 69 6F 6E 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C 05 00 00 00 00 00 00 00 66 61 6C 73 65 DE AD DE AD\r\n    Size  = 851\r\n\r\n\r\n  Size  = 877\r\n\r\nTable 'SubscriptionsRuntimeState':\r\n\r\n  Key 'mgmt://canary/subscription':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 42 00 00 00 00 00 00 00 0A 00 00 00 00 00 00 00 22 72 63 3A 45 6D 70 74 79 22 01 00 00 00 00 00 00 00 31 01 00 00 00 00 00 00 00 30 01 00 00 00 00 00 00 00 30 01 00 00 00 00 00 00 00 30 04 00 00 00 00 00 00 00 74 72 75 65 DE AD DE AD\r\n    Size  = 173\r\n\r\n\r\n  Size  = 223\r\n\r\nTotal size = 2611\r\n\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "Console.WriteLine(store.DebugView);"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Define artifacts\n",
        "\n",
        "Illustrates populating the registry of defined artifacts in the engine. This is a one-time step for the environment creating a new engine.\n",
        "\n",
        "* Artifact types that are defined include:\n",
        "  * Observables, e.g. sources of events, or query operators.\n",
        "  * Observers, e.g. sinks for events, or event handlers.\n",
        "  * Stream factories, not shown here. Useful for creation of \"subjects\" local to the engine.\n",
        "  * Subscription factories, not shown here. Useful for \"templates\" to create subscriptions with parameters.\n",
        "* All Reactor artifacts use URIs for naming purposes.\n",
        "\n",
        "The key take-away is that Reactor engines are empty by default and have no built-in artifacts whatsoever. The environment controls the registry, which includes standard query operators, specialized query operators, etc.\n",
        "\n",
        "> **Note:** There's an alternative approach to having artifacts defined in and persisted by individual engine instances. The engine can also be parameterized on a queryable external catalog. This is useful for homogeneous environments."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    await ctx.DefineObserverAsync(new Uri(\"iot://reactor/observers/cout\"), ctx.Provider.CreateQbserver<T>(Expression.New(typeof(ConsoleObserver<T>))), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<TimeSpan, DateTimeOffset>(new Uri(\"iot://reactor/observables/timer\"), t => new TimerObservable(t).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Once more, let's print the checkpoint store. This time we'll see entries for the two artifacts defined above."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "Table 'TxMetadata':\r\n\r\n  Key 'Latest':\r\n    Value = 01 33\r\n    Size  = 14\r\n\r\n  Key 'ActiveCount':\r\n    Value = 01 31\r\n    Size  = 24\r\n\r\n  Key 'HeldCount':\r\n    Value = 01 31\r\n    Size  = 20\r\n\r\n\r\n  Size  = 78\r\n\r\nTable 'Observers':\r\n\r\n  Key 'mgmt://canary/observer':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 88 01 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 4D 60 22 2C 30 2C 22 4E 6F 70 22 2C 31 2C 5B 5D 2C 32 5D 2C 5B 22 4D 3C 3E 22 2C 30 2C 5B 33 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 4F 62 73 65 72 76 65 72 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 4F 62 73 65 72 76 65 72 60 31 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 2D 31 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 31 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 52 65 61 71 74 69 76 65 2E 43 6F 72 65 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 2E 28 29 22 2C 31 2C 5B 5D 5D 7D 18 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 65 72 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 649\r\n\r\n  Key 'iot://reactor/observers/cout':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 72 01 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 43 22 2C 32 2C 5B 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 52 65 61 71 74 6F 72 2E 49 6F 54 2E 43 6F 6E 73 6F 6C 65 4F 62 73 65 72 76 65 72 60 31 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 4C 69 6E 71 2E 43 6F 6D 70 69 6C 65 72 53 65 72 76 69 63 65 73 2E 54 79 70 65 53 79 73 74 65 6D 2E 54 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 30 2C 5B 31 5D 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 52 65 61 71 74 6F 72 2E 49 6F 54 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 4E 75 71 6C 65 6F 6E 2E 4C 69 6E 71 2E 43 6F 6D 70 69 6C 65 72 53 65 72 76 69 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 6E 65 77 22 2C 30 2C 5B 5D 5D 7D 1E 00 00 00 00 00 00 00 22 69 6F 74 3A 2F 2F 72 65 61 63 74 6F 72 2F 6F 62 73 65 72 76 65 72 73 2F 63 6F 75 74 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 645\r\n\r\n\r\n  Size  = 1312\r\n\r\nTable 'Observables':\r\n\r\n  Key 'mgmt://canary/observable':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 E1 01 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 4D 60 22 2C 30 2C 22 45 6D 70 74 79 22 2C 31 2C 5B 5D 2C 32 5D 2C 5B 22 4D 3C 3E 22 2C 30 2C 5B 33 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 53 75 62 73 63 72 69 62 61 62 6C 65 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 62 61 62 6C 65 60 31 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 2D 31 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 32 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 52 65 61 71 74 69 76 65 2E 4C 69 6E 71 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 52 65 61 71 74 69 76 65 2E 49 6E 74 65 72 66 61 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 2E 28 29 22 2C 31 2C 5B 5D 5D 7D 1A 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 61 62 6C 65 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 744\r\n\r\n  Key 'iot://reactor/observables/timer':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 3C 02 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 4D 65 6D 62 65 72 73 22 3A 5B 5B 22 43 22 2C 31 2C 5B 30 5D 5D 5D 2C 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 54 69 6D 65 53 70 61 6E 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 6F 72 2E 49 6F 54 2E 54 69 6D 65 72 4F 62 73 65 72 76 61 62 6C 65 22 2C 31 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 62 61 62 6C 65 60 31 22 2C 32 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 44 61 74 65 54 69 6D 65 4F 66 66 73 65 74 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 32 2C 5B 33 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 46 75 6E 63 60 32 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 35 2C 5B 30 2C 34 5D 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 2C 22 52 65 61 71 74 6F 72 2E 49 6F 54 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 2C 22 52 65 61 71 74 69 76 65 2E 49 6E 74 65 72 66 61 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 3D 3E 22 2C 36 2C 5B 22 3C 3A 22 2C 34 2C 5B 22 6E 65 77 22 2C 30 2C 5B 5B 22 24 22 2C 30 2C 30 5D 5D 5D 5D 2C 5B 5B 30 2C 22 70 30 22 5D 5D 5D 7D 21 00 00 00 00 00 00 00 22 69 6F 74 3A 2F 2F 72 65 61 63 74 6F 72 2F 6F 62 73 65 72 76 61 62 6C 65 73 2F 74 69 6D 65 72 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C DE AD DE AD\r\n    Size  = 856\r\n\r\n\r\n  Size  = 1622\r\n\r\nTable 'Subscriptions':\r\n\r\n  Key 'mgmt://canary/subscription':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 39 02 00 00 00 00 00 00 7B 22 43 6F 6E 74 65 78 74 22 3A 7B 22 54 79 70 65 73 22 3A 5B 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 46 75 6E 63 60 33 22 2C 30 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 62 61 62 6C 65 60 31 22 2C 31 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 6E 74 33 32 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 31 2C 5B 32 5D 5D 2C 5B 22 3A 3A 22 2C 22 53 79 73 74 65 6D 2E 49 4F 62 73 65 72 76 65 72 60 31 22 2C 30 5D 2C 5B 22 3C 3E 22 2C 34 2C 5B 32 5D 5D 2C 5B 22 3A 3A 22 2C 22 52 65 61 71 74 69 76 65 2E 49 53 75 62 73 63 72 69 70 74 69 6F 6E 22 2C 31 5D 2C 5B 22 3C 3E 22 2C 30 2C 5B 33 2C 35 2C 36 5D 5D 5D 2C 22 41 73 73 65 6D 62 6C 69 65 73 22 3A 5B 22 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 2C 22 52 65 61 71 74 69 76 65 2E 49 6E 74 65 72 66 61 63 65 73 2C 20 56 65 72 73 69 6F 6E 3D 31 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 6E 75 6C 6C 22 5D 2C 22 56 65 72 73 69 6F 6E 22 3A 22 30 2E 39 2E 30 2E 30 22 2C 22 47 6C 6F 62 61 6C 73 22 3A 5B 5B 37 2C 22 72 78 3A 2F 2F 6F 62 73 65 72 76 61 62 6C 65 2F 73 75 62 73 63 72 69 62 65 22 5D 2C 5B 33 2C 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 61 62 6C 65 22 5D 2C 5B 35 2C 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 6F 62 73 65 72 76 65 72 22 5D 5D 7D 2C 22 45 78 70 72 65 73 73 69 6F 6E 22 3A 5B 22 28 29 22 2C 5B 22 24 22 2C 30 5D 2C 5B 5B 22 24 22 2C 31 5D 2C 5B 22 24 22 2C 32 5D 5D 5D 7D 1C 00 00 00 00 00 00 00 22 6D 67 6D 74 3A 2F 2F 63 61 6E 61 72 79 2F 73 75 62 73 63 72 69 70 74 69 6F 6E 22 6A 00 00 00 00 00 00 00 22 53 79 73 74 65 6D 2E 4F 62 6A 65 63 74 2C 20 53 79 73 74 65 6D 2E 50 72 69 76 61 74 65 2E 43 6F 72 65 4C 69 62 2C 20 56 65 72 73 69 6F 6E 3D 35 2E 30 2E 30 2E 30 2C 20 43 75 6C 74 75 72 65 3D 6E 65 75 74 72 61 6C 2C 20 50 75 62 6C 69 63 4B 65 79 54 6F 6B 65 6E 3D 37 63 65 63 38 35 64 37 62 65 61 37 37 39 38 65 22 04 00 00 00 00 00 00 00 6E 75 6C 6C 05 00 00 00 00 00 00 00 66 61 6C 73 65 DE AD DE AD\r\n    Size  = 851\r\n\r\n\r\n  Size  = 877\r\n\r\nTable 'SubscriptionsRuntimeState':\r\n\r\n  Key 'mgmt://canary/subscription':\r\n    Value = 42 44 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 04 4A 53 4F 4E 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 42 00 00 00 00 00 00 00 0A 00 00 00 00 00 00 00 22 72 63 3A 45 6D 70 74 79 22 01 00 00 00 00 00 00 00 31 01 00 00 00 00 00 00 00 30 01 00 00 00 00 00 00 00 30 01 00 00 00 00 00 00 00 30 04 00 00 00 00 00 00 00 74 72 75 65 DE AD DE AD\r\n    Size  = 173\r\n\r\n\r\n  Size  = 223\r\n\r\nTable '2TxObservers':\r\n\r\n\r\n  Size  = 24\r\n\r\nTable '2TxObservables':\r\n\r\n\r\n  Size  = 28\r\n\r\nTotal size = 4164\r\n\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "Console.WriteLine(store.DebugView);"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create a subscription\n",
        "\n",
        "Illustrates the user programming surface of Reactor. The environment should provide an `IReactiveProxy` \"context\" object to the user. It provides an API similar to LINQ to SQL and Rx:\n",
        "\n",
        "* `Get*` to obtain artifacts, using their well-known URIs.\n",
        "* Compose queries over those artifacts.\n",
        "* Submit them to the engine using async operations (e.g. `SubscribeAsync` in lieu of `Subscribe` in Rx).\n",
        "\n",
        "The sample below shows the most basic subscription, merely connecting a source (observable) and a sink (observer). The resulting subscription has a name, which can be used to delete it later.\n",
        "\n",
        "A few notes on `IReactiveProxy`:\n",
        "\n",
        "* APIs are asynchronous to cover I/O:\n",
        "  * In a distributed service, this includes submitting a serialized expression tree across machine boundaries.\n",
        "  * At the engine level (shown here), this includes the transaction log operation for the create operation (enabling replay of the creation operation in the event of engine failure before the next checkpoint).\n",
        "* The `ReactorContext` type shown below implements this interface:\n",
        "  * Think of it being analogous to DataContext in LINQ to SQL, which has methods like `GetTable<T>`.\n",
        "  * Derived types can provide friendly accessors for artifacts, just like LINQ to SQL could have a `NorthwindDataContext` providing a `Products` property to hide `GetTable<Product>(\"Product\")`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:18 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:19 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:20 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:21 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:22 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:23 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var timer = ctx.GetObservable<TimeSpan, DateTimeOffset>(new Uri(\"iot://reactor/observables/timer\"));\n",
        "    var cout = ctx.GetObserver<DateTimeOffset>(new Uri(\"iot://reactor/observers/cout\"));\n",
        "\n",
        "    await timer(TimeSpan.FromSeconds(1)).SubscribeAsync(cout, new Uri(\"iot://reactor/subscriptions/heartbeat\"), null, CancellationToken.None);\n",
        "\n",
        "    // Let's wait a little so we can see some output before shutting down the engine.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Recover the engine\n",
        "\n",
        "Illustrates checkpoint/recovery. The subscription is running again after failover."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:24 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:25 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:26 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:27 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:28 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:29 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    // Let's wait a little so we can see some output before shutting down the engine.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Dispose a subscription\n",
        "\n",
        "Illustrates user code to dispose an existing subscription.\n",
        "\n",
        "See remarks on `IReactiveProxy` higher up. The pattern is identical:\n",
        "\n",
        "- Use a `Get` operation to get a proxy to the artifact, in this case a subscription, using the URI.\n",
        "- Invoke an asynchronous operation to act on it, in this case DisposeAsync to dispose the subscription.\n",
        "\n",
        "The asynchronous nature of the operation is again due to:\n",
        "\n",
        "- The ability to send the operation across machine boundaries.\n",
        "- The transaction log in the engine to persist the deletion operation in the event of failure before the next checkpoint."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:31 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:32 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:33 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:34 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:35 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:36 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Disposing the subscription...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Disposed the subscription...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    // Let's wait a little so we can see some output before disposing the subscription.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var heartbeat = ctx.GetSubscription(new Uri(\"iot://reactor/subscriptions/heartbeat\"));\n",
        "\n",
        "    Console.WriteLine(\"Disposing the subscription...\");\n",
        "\n",
        "    await heartbeat.DisposeAsync();\n",
        "\n",
        "    Console.WriteLine(\"Disposed the subscription...\");\n",
        "\n",
        "    // Let's wait a little so we can confirm that the subscription has stopped running.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Define query operators\n",
        "\n",
        "Illustration of defining query operators, similar to defining other artifacts higher up. A few remarks:\n",
        "\n",
        "- No operators are built-in. Below, we define essential operators like `Where`, `Select`, and `Take`. The URI for these is not even prescribed; the environment picks those.\n",
        "- Implementations of the operators are provided in `Reaqtive`, similar to `System.Reactive` for classic Rx. The difference is mainly due to support for state persistence, which classic Rx lacks.\n",
        "- Custom operators are as first-class as \"standard query operators\". That is, the query engine does not have an opinion about the operator surface provided.\n",
        "\n",
        "Some ugly technicalities show up below, but those are entirely irrelevant to the user experience. The code below is part of the one-time setup provided by the environment. In particular:\n",
        "\n",
        "- Define operations are done through `IReactiveProxy`, but could also be done straight on the engine (though it brings some additional complexity when doing so).\n",
        "- There's some conversion friction to build expressions that fit through a \"queryable\" expression-tree based API but eventually bind to types in Reaqtive. That's all the As* stuff below."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, bool>, T>(new Uri(\"iot://reactor/observables/filter\"), (source, predicate) => source.AsSubscribable().Where(predicate).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int, bool>, T>(new Uri(\"iot://reactor/observables/filter/indexed\"), (source, predicate) => source.AsSubscribable().Where(predicate).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, R>, R>(new Uri(\"iot://reactor/observables/map\"), (source, selector) => source.AsSubscribable().Select(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int, R>, R>(new Uri(\"iot://reactor/observables/map/indexed\"), (source, selector) => source.AsSubscribable().Select(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, T>(new Uri(\"iot://reactor/observables/take\"), (source, count) => source.AsSubscribable().Take(count).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustration of a more sophisticated query expression written by the user, using the operators defined above.\n",
        "\n",
        "Note that all Rx operators can be defined and used with Reactor, so this merely serves as an example of a select subset of those.\n",
        "\n",
        "The fluent experience using extension methods (and hence supporting query expression \"LINQ\" syntax as well) is introduced through the Operators type with method definitions like this:\n",
        "\n",
        "```csharp\n",
        "[KnownResource(\"iot://reactor/observables/filter\")]\n",
        "public static IAsyncReactiveQbservable<T> Where<T>(this IAsyncReactiveQbservable<T> source, Expression<Func<T, bool>> predicate)\n",
        "```\n",
        "\n",
        "This is part of the APIs provided by the environment.\n",
        "\n",
        "Note the use of `KnownResource` to refer to the URI of the defined artifact. Any (extension) method besides the standard query operators can use this mechanism to allow fluent formulation of queries.\n",
        "This again serves to show that nothing is built-in in Reactor: Where, Select, etc. aren't any more special than any other \"non-standard\" query operator.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:40 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:41 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:42 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:43 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:44 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:45 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var timer = ctx.GetObservable<TimeSpan, DateTimeOffset>(new Uri(\"iot://reactor/observables/timer\"));\n",
        "    var cout = ctx.GetObserver<string>(new Uri(\"iot://reactor/observers/cout\"));\n",
        "\n",
        "    var res = timer(TimeSpan.FromSeconds(0.5)).Where((x, i) => i % 2 == 0).Select(dt => dt.ToString()).Take(8);\n",
        "\n",
        "    await res.SubscribeAsync(cout, new Uri(\"iot://reactor/subscriptions/heartbeat/advanced\"), null, CancellationToken.None);\n",
        "\n",
        "    // Let's wait a little so we can see some output before shutting down the engine.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustrates recovery of a stateful query. The query above has a `Take(8)`, so part of the persistence includes the remaining event count."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:48 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnNext(3/5/2021 9:07:49 PM -08:00)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "OnCompleted()\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    // Let's wait a little so we can see some output before shutting down the engine.\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Metadata queries\n",
        "\n",
        "Illustration of metadata queries on the engine.\n",
        "\n",
        "`IReactiveProxy` exposes queryable collections such as Observables, Observers, Subscriptions that can be used to enumerate artifacts in the engine, or to formulate queries.\n",
        "Note that LINQ query provider support is limited in the engine today, but the registry can be indexed efficiently (e.g. `ContainsKey`, `SingleOrDefault`), and can be enumerated.\n",
        "Work on metadata queries in the engine has been hampered by the lack of `IAsyncQueryable<T>` support (which is only coming to .NET now, over 5 years later). We could go back to add rich querying support if such a need arises.\n",
        "\n",
        "For the IoT environment, a `ContainsKey` query could be useful to check whether a query has already been defined, and even to obain its expression tree, e.g. if we wish to do some idempotent `Create` operation."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "Found subscription: True\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "IoT operators defined in engine:\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/map/indexed\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/map\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/take\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/filter/indexed\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/filter\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "  iot://reactor/observables/timer\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var found = ctx.Subscriptions.ContainsKey(new Uri(\"iot://reactor/subscriptions/heartbeat/advanced\"));\n",
        "    Console.WriteLine(\"Found subscription: \" + found);\n",
        "    Console.WriteLine();\n",
        "\n",
        "    Console.WriteLine(\"IoT operators defined in engine:\");\n",
        "    foreach (var observable in ctx.Observables.AsEnumerable().Where(kv => kv.Key.Scheme == \"iot\"))\n",
        "    {\n",
        "        Console.WriteLine(\"  \" + observable.Key);\n",
        "    }\n",
        "\n",
        "    await Task.Yield();\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustration of disposing a subscription, again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var heartbeat = ctx.GetSubscription(new Uri(\"iot://reactor/subscriptions/heartbeat/advanced\"));\n",
        "    await heartbeat.DisposeAsync();\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Ingress and egress\n",
        "\n",
        "Illustration of defining ingress/egress proxies as observable/observer artifacts.\n",
        "\n",
        "Also see the implementation of `IngressObservable<T>` and `EgressObserver<T>`, which use the ingress/egress manager to connect to the outside world. The essence is this:\n",
        "\n",
        "- To the query running inside the engine, these look like ordinary Rx artifacts implemented using interfaces base classes provided by Reactor:\n",
        "  - `ISubscribable<T>` rather than `IObservable<T>`, to support the richer lifecycle of artifacts in Reactor compared to Rx.\n",
        "  - `Load`/`Save` state operations for checkpointing.\n",
        "- The external world communicates with the engine using a variant of the observable/observer interfaces, namely `IReliable*<T>`:\n",
        "  - Events received and produced have sequence numbers.\n",
        "  - Subscription handles to receive events from the outside world have additional operations:\n",
        "    - `Start(long)` to replay events from the given sequence number.\n",
        "    - `AcknowledgeRange(long)` to allow the external service to (optionally) prune events that are no longer needed by the engine.\n",
        "- Proxies in the engine use the sequence number to provide reliability:\n",
        "  - `Save` persists the latest received sequence number. `Load` gets it back.\n",
        "  - Upon restart of an ingress proxy, the restored sequence number is used to ask for replay of events.\n",
        "  - Upon a successful checkpoint, the latest received sequence number is acknowledged to the source (allowing pruning).\n",
        "\n",
        "The Reactor service implements such ingress/egress mechanisms using services like EventHub."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    await ctx.DefineObserverAsync<string, T>(new Uri(\"iot://reactor/observers/egress\"), stream => new EgressObserver<T>(stream).AsAsyncQbserver(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<string, T>(new Uri(\"iot://reactor/observables/ingress\"), stream => new IngressObservable<T>(stream).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Mimic the outside world by creating streams in the ingress/egress manager. In reality, this would come from the environment, e.g. sensor data.\n",
        "\n",
        "We create two streams:\n",
        "\n",
        "- `bar` will be used to receive events from the outside world and to perform event processing queries on those events;\n",
        "- `foo` will be used to send events produced by the queries to the outside world.\n",
        "\n",
        "Thus:\n",
        "\n",
        "- the query in the engine will subscribe to bar and emit events into foo;\n",
        "- the outside world will emit events into bar and subscribe to foo.\n",
        "\n",
        "Also note that all events outside the engine boundaries have sequence numbers."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "Setting up external streams...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "Console.WriteLine(\"Setting up external streams...\");\n",
        "\n",
        "var bar = iemgr.CreateSubject<int>(\"bar\");\n",
        "var foo = iemgr.CreateSubject<int>(\"foo\");"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Now we can create some simulated event producer."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "Task ProduceBarEvents(int start, bool log, CancellationToken token)\n",
        "{\n",
        "    return Task.Run(async () =>\n",
        "    {\n",
        "        for (int i = start; !token.IsCancellationRequested; i++)\n",
        "        {\n",
        "            var e = (i, 10 * i);\n",
        "\n",
        "            if (log)\n",
        "            {\n",
        "                Console.WriteLine(\"bar> \" + e);\n",
        "            }\n",
        "\n",
        "            bar.OnNext(e);\n",
        "\n",
        "            await Task.Delay(TimeSpan.FromSeconds(1));\n",
        "        }\n",
        "    });\n",
        "}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustrates a simple pass-through query where events are received from external stream `bar` and forwarded to external stream `foo`.\n",
        "\n",
        "Note that the observable and observer proxies are parameterized on the external stream name. This can obviously be hidden in a number of ways:\n",
        "\n",
        "- `Define` non-parameterized observable and observer artifacts for input/output streams, using a descriptive URI, e.g. `iot://sensor/temperature`.\n",
        "- `Create` a context derived from `ReactorContext` that provides properties that provide direct access to those, e.g. `ctx.Temperature`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "bar> (10, 100)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (0, 100)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "bar> (11, 110)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (1, 110)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "bar> (12, 120)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (2, 120)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "bar> (13, 130)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (3, 130)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "bar> (14, 140)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (4, 140)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "bar> (15, 150)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (5, 150)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Engine unloaded.\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "var observer =\n",
        "    Observer.Create<(long sequenceId, int value)>(\n",
        "        x => Console.WriteLine(\"foo> \" + x),\n",
        "        ex => Console.WriteLine(\"foo> \" + ex.Message),\n",
        "        () => Console.WriteLine(\"foo> Done\")\n",
        "    );\n",
        "\n",
        "var fooLogger = foo.Subscribe(observer);\n",
        "\n",
        "var stopBarProducer = new CancellationTokenSource();\n",
        "stopBarProducer.CancelAfter(TimeSpan.FromSeconds(30));\n",
        "var producer = ProduceBarEvents(start: 10, log: true, stopBarProducer.Token);\n",
        "\n",
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var input = ctx.GetObservable<string, int>(new Uri(\"iot://reactor/observables/ingress\"));\n",
        "    var output = ctx.GetObserver<string, int>(new Uri(\"iot://reactor/observers/egress\"));\n",
        "\n",
        "    await input(\"bar\").SubscribeAsync(output(\"foo\"), new Uri(\"iot://reactor/subscriptions/in_out\"), null, CancellationToken.None);\n",
        "\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});\n",
        "\n",
        "Console.WriteLine(\"Engine unloaded.\");"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Note how the sequence numbers on the input and output do not match. We started producing inputs at some sequence number with value `10`. The outputs are totally independent and have their own sequence number."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustrates the replay behavior in the face of failover. During the downtime between running the previous cell and running the next cell, events were produced (note we set the event producer for `bar` to stop after `30` seconds). Upon recovery, these events are replayed."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "foo> (6, 150)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (7, 160)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (8, 170)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (9, 180)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (10, 190)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (11, 200)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (12, 210)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (13, 220)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (14, 230)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (15, 240)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (16, 250)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (17, 260)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (18, 270)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (19, 280)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (20, 290)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (21, 300)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (22, 310)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (23, 320)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (24, 330)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (25, 340)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (26, 350)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (27, 360)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (28, 370)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (29, 380)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "foo> (30, 390)\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    // Let the engine run for a bit to see the replay in action.\n",
        "    await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Illustration of disposing a subscription, again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var in_out = ctx.GetSubscription(new Uri(\"iot://reactor/subscriptions/in_out\"));\n",
        "    await in_out.DisposeAsync(CancellationToken.None);\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Define more query operators\n",
        "\n",
        "Illustrates the definition of higher-order operators such as SelectMany and GroupBy which operate on sequences of sequences (IObservable<IObservable<T>>) which is one of the most powerful aspects of Rx."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    // Average\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<int>, double>(new Uri(\"iot://reactor/observables/average/int32\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<long>, double>(new Uri(\"iot://reactor/observables/average/int64\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<double>, double>(new Uri(\"iot://reactor/observables/average/double\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int>, double>(new Uri(\"iot://reactor/observables/average/selector/int32\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, long>, double>(new Uri(\"iot://reactor/observables/average/selector/int64\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, double>, double>(new Uri(\"iot://reactor/observables/average/selector/double\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "\n",
        "    // DistinctUntilChanged\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, T>(new Uri(\"iot://reactor/observables/distinct\"), source => source.AsSubscribable().DistinctUntilChanged().AsAsyncQbservable(), null, CancellationToken.None);\n",
        "\n",
        "    // SelectMany\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, ISubscribable<R>>, R>(new Uri(\"iot://reactor/observables/bind\"), (source, selector) => source.AsSubscribable().SelectMany(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "\n",
        "    // Window\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/hopping/time\"), (source, duration) => source.AsSubscribable().Window(duration).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/hopping/count\"), (source, count) => source.AsSubscribable().Window(count).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, TimeSpan, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/sliding/time\"), (source, duration, shift) => source.AsSubscribable().Window(duration, shift).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/sliding/count\"), (source, count, skip) => source.AsSubscribable().Window(count, skip).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/ferry\"), (source, duration, count) => source.AsSubscribable().Window(duration, count).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "\n",
        "    // GroupBy\n",
        "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, R>, IGroupedSubscribable<R, T>>(new Uri(\"iot://reactor/observables/group\"), (source, selector) => source.AsSubscribable().GroupBy(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
        "});"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Entity types\n",
        "\n",
        "Reactor Core is built to be flexible with regards to data models, but the default data model that's well-supported originates from a graph database effort in Bing that predates Reactor. The `[Mapping]` attributes below are the means to annotate properties. These property names are used to normalize entity types in the serialized expression representation, so the query is not dependent on a concrete type in an assembly, thus allowing the structure of data types (here to represent events) to be serialized across machine boundaries without deployment of binaries."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "public class SensorReading\n",
        "{\n",
        "    [Mapping(\"iot://sensor/reading/room\")]\n",
        "    public string Room { get; set; }\n",
        "\n",
        "    [Mapping(\"iot://sensor/reading/temperature\")]\n",
        "    public double Temperature { get; set; }\n",
        "}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## A temperature simulator\n",
        "\n",
        "Add other streams to connect to the environment, simulating a temperature sensor reading and a feedback channel to control an A/C unit."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "var readings = iemgr.CreateSubject<SensorReading>(\"bart://sensors/home/livingroom/temperature/readings\");\n",
        "var settings = iemgr.CreateSubject<double?>(\"bart://sensors/home/livingroom/temperature/settings\");"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Next, we define a few constants for the simulation."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "var rand = new Random();\n",
        "\n",
        "//\n",
        "// Speed and granularity of simulation.\n",
        "//\n",
        "var timeStep = TimeSpan.FromMinutes(15);\n",
        "var simulationDelay = TimeSpan.FromMilliseconds(250);\n",
        "\n",
        "//\n",
        "// Absolute value of temperature gain/loss per unit time of the house adjusting to the outside temperature.\n",
        "//\n",
        "var insulationTemperatureIncrement = 0.1;\n",
        "\n",
        "//\n",
        "// Absolute value of temperature gain/loss per unit time due to the A/C unit cooling down or heating up.\n",
        "//\n",
        "var acTemperatureIncrement = 0.2;\n",
        "\n",
        "//\n",
        "// Temperature sensitivity of the thermostat to trigger turning off the A/C unit, i.e. within this range from target.\n",
        "//\n",
        "var thermostatSensitivity = 0.5;\n",
        "\n",
        "//\n",
        "// Configuration of simulation: minimum and maximum temperature outside, and coldest time of day.\n",
        "//\n",
        "var outsideMin = 55;\n",
        "var outsideMax = 85;\n",
        "var coldestTime = new TimeSpan(5, 0, 0); // 5AM\n",
        "\n",
        "//\n",
        "// Scale for the temperature range, to multiply [0..1] by to obtain a temperature value that can be added to the minimum.\n",
        "//\n",
        "var scale = outsideMax - outsideMin;\n",
        "\n",
        "//\n",
        "// Offset to the midpoint of the temperature range. Outside temperature will vary as a sine wave around this value.\n",
        "//\n",
        "var offset = outsideMin + scale / 2;\n",
        "\n",
        "#pragma warning disable CA5394 // Do not use insecure randomness. (Okay for simulation purposes.)\n",
        "\n",
        "//\n",
        "// Random initial value inside, within the range of temperatures.\n",
        "//\n",
        "var inside = outsideMin + rand.NextDouble() * scale;\n",
        "\n",
        "#pragma warning restore CA5394\n",
        "\n",
        "//\n",
        "// null if A/C unit is off; otherwise, target temperature.\n",
        "//\n",
        "var target = default(double?);\n",
        "\n",
        "//\n",
        "// Clock driven by the simulation.\n",
        "//\n",
        "var time = DateTime.Today;"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Now we can write a simulator routine that will generate `reading`, and set up a subscription to the `settings` stream to show the results emitted by the Reaqtor query we'll construct later."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [],
      "source": [
        "IDisposable SubscribeToSettingsStream()\n",
        "{\n",
        "    //\n",
        "    // Print commands arriving at thermostat.\n",
        "    //\n",
        "    return settings.Subscribe(Observer.Create<(long sequenceId, double? item)>(s =>\n",
        "    {\n",
        "        target = s.item;\n",
        "        Console.WriteLine($\"{time} thermostat> {(target == null ? \"OFF\" : \"ON \" + (target > inside ? \"heating\" : \"cooling\") + \" to \" + target)}\");\n",
        "    }));\n",
        "}\n",
        "\n",
        "Task RunReadingsGenerator(CancellationToken token)\n",
        "{\n",
        "    //\n",
        "    // Run simulation which adjusts both inside and outside temperature.\n",
        "    //\n",
        "    return Task.Run(async () =>\n",
        "    {\n",
        "        while (!token.IsCancellationRequested)\n",
        "        {\n",
        "            var now = (time.TimeOfDay - coldestTime - TimeSpan.FromHours(6)).TotalSeconds;\n",
        "            var secondsPerDay = TimeSpan.FromHours(24).TotalSeconds;\n",
        "\n",
        "            var outside = scale * Math.Sin(2 * Math.PI * now / secondsPerDay) / 2 + offset;\n",
        "\n",
        "            var environmentEffect = outside < inside ? -insulationTemperatureIncrement : insulationTemperatureIncrement;\n",
        "            var acUnitEffect = target != null ? (target < inside ? -acTemperatureIncrement : acTemperatureIncrement) : 0.0;\n",
        "\n",
        "            inside += environmentEffect + acUnitEffect;\n",
        "\n",
        "            if (target != null && Math.Abs(target.Value - inside) < thermostatSensitivity)\n",
        "            {\n",
        "                target = null;\n",
        "            }\n",
        "\n",
        "            Console.WriteLine($\"{time} temperature> inside = {inside} outside = {outside} target = {target}\");\n",
        "            readings.OnNext((Environment.TickCount, new SensorReading { Room = \"Hallway\", Temperature = inside }));\n",
        "\n",
        "            await Task.Delay(simulationDelay);\n",
        "            time += timeStep;\n",
        "        }\n",
        "    });\n",
        "}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "In the next well, we'll write a higher-order query using `Window` and `SelectMany`, and run the simulator and logger while the query is running."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "dotnet_interactive": {
          "language": "csharp"
        }
      },
      "outputs": [
        {
          "data": {
            "text/plain": "Starting simulator for temperature sensor readings...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Setting up query engine...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:00:00 AM temperature> inside = 76.02714887649154 outside = 66.11771432346218 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Creating subscription...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:15:00 AM temperature> inside = 75.92714887649154 outside = 65.17840802045258 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:30:00 AM temperature> inside = 75.82714887649155 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:45:00 AM temperature> inside = 75.72714887649155 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:45:00 AM thermostat> OFF\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:00:00 AM temperature> inside = 75.62714887649156 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:15:00 AM temperature> inside = 75.52714887649157 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:30:00 AM temperature> inside = 75.42714887649157 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:45:00 AM temperature> inside = 75.32714887649158 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:00:00 AM temperature> inside = 75.22714887649158 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:15:00 AM temperature> inside = 75.12714887649159 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:30:00 AM temperature> inside = 75.0271488764916 outside = 58.09969989563147 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:45:00 AM temperature> inside = 74.9271488764916 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:00:00 AM temperature> inside = 74.8271488764916 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:15:00 AM temperature> inside = 74.72714887649161 outside = 56.54690887700967 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:30:00 AM temperature> inside = 74.62714887649162 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:45:00 AM temperature> inside = 74.52714887649162 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:00:00 AM temperature> inside = 74.42714887649163 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:15:00 AM temperature> inside = 74.32714887649163 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:30:00 AM temperature> inside = 74.22714887649164 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:45:00 AM temperature> inside = 74.12714887649165 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:00:00 AM temperature> inside = 74.02714887649165 outside = 55 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:15:00 AM temperature> inside = 73.92714887649166 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:30:00 AM temperature> inside = 73.82714887649166 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:45:00 AM temperature> inside = 73.72714887649167 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:00:00 AM temperature> inside = 73.62714887649167 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:15:00 AM temperature> inside = 73.52714887649168 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:30:00 AM temperature> inside = 73.42714887649169 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:45:00 AM temperature> inside = 73.32714887649169 outside = 56.54690887700968 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:00:00 AM temperature> inside = 73.2271488764917 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:15:00 AM temperature> inside = 73.1271488764917 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:30:00 AM temperature> inside = 73.02714887649171 outside = 58.099699895631474 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:45:00 AM temperature> inside = 72.92714887649171 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:00:00 AM temperature> inside = 72.82714887649172 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:15:00 AM temperature> inside = 72.72714887649173 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:30:00 AM temperature> inside = 72.62714887649173 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:45:00 AM temperature> inside = 72.52714887649174 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:00:00 AM temperature> inside = 72.42714887649174 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:15:00 AM temperature> inside = 72.32714887649175 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:30:00 AM temperature> inside = 72.22714887649175 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:45:00 AM temperature> inside = 72.12714887649176 outside = 65.17840802045258 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:00:00 AM temperature> inside = 72.02714887649176 outside = 66.1177143234622 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:15:00 AM temperature> inside = 71.92714887649177 outside = 67.07364516975808 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:30:00 AM temperature> inside = 71.82714887649178 outside = 68.04210711669923 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:45:00 AM temperature> inside = 71.72714887649178 outside = 69.01895306154785 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:00:00 AM temperature> inside = 71.62714887649179 outside = 70 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:15:00 AM temperature> inside = 71.5271488764918 outside = 70.98104693845215 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:30:00 AM temperature> inside = 71.62714887649179 outside = 71.95789288330077 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:45:00 AM temperature> inside = 71.72714887649178 outside = 72.92635483024192 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:00:00 PM temperature> inside = 71.82714887649178 outside = 73.8822856765378 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:15:00 PM temperature> inside = 71.92714887649177 outside = 74.82159197954742 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:30:00 PM temperature> inside = 72.02714887649176 outside = 75.74025148547635 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 12:45:00 PM temperature> inside = 72.12714887649176 outside = 76.63433035328502 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:00:00 PM temperature> inside = 72.22714887649175 outside = 77.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:15:00 PM temperature> inside = 72.32714887649175 outside = 78.33355349529404 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:30:00 PM temperature> inside = 72.42714887649174 outside = 79.1314214351308 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 1:45:00 PM temperature> inside = 72.52714887649174 outside = 79.89018722650103 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:00:00 PM temperature> inside = 72.62714887649173 outside = 80.60660171779821 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:15:00 PM temperature> inside = 72.72714887649173 outside = 81.27759711218467 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:30:00 PM temperature> inside = 72.82714887649172 outside = 81.90030010436853 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 2:45:00 PM temperature> inside = 72.92714887649171 outside = 82.47204418453818 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:00:00 PM temperature> inside = 73.02714887649171 outside = 82.99038105676658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:15:00 PM temperature> inside = 73.1271488764917 outside = 83.45309112299033 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:30:00 PM temperature> inside = 73.2271488764917 outside = 83.8581929876693 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 3:45:00 PM temperature> inside = 73.32714887649169 outside = 84.20395194242658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:00:00 PM temperature> inside = 73.42714887649169 outside = 84.48888739433602 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:15:00 PM temperature> inside = 73.52714887649168 outside = 84.71177920604846 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:30:00 PM temperature> inside = 73.62714887649167 outside = 84.87167292060715 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 4:45:00 PM temperature> inside = 73.72714887649167 outside = 84.96788384857905 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:00:00 PM temperature> inside = 73.82714887649166 outside = 85 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:15:00 PM temperature> inside = 73.92714887649166 outside = 84.96788384857905 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:30:00 PM temperature> inside = 74.02714887649165 outside = 84.87167292060715 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 5:45:00 PM temperature> inside = 74.12714887649165 outside = 84.71177920604846 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:00:00 PM temperature> inside = 74.22714887649164 outside = 84.48888739433602 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:15:00 PM temperature> inside = 74.32714887649163 outside = 84.20395194242658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:30:00 PM temperature> inside = 74.42714887649163 outside = 83.8581929876693 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 6:45:00 PM temperature> inside = 74.52714887649162 outside = 83.45309112299033 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:00:00 PM temperature> inside = 74.62714887649162 outside = 82.99038105676658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:15:00 PM temperature> inside = 74.72714887649161 outside = 82.47204418453818 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:30:00 PM temperature> inside = 74.8271488764916 outside = 81.90030010436853 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 7:45:00 PM temperature> inside = 74.9271488764916 outside = 81.27759711218467 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:00:00 PM temperature> inside = 75.0271488764916 outside = 80.60660171779821 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:15:00 PM temperature> inside = 75.12714887649159 outside = 79.89018722650104 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:30:00 PM temperature> inside = 75.22714887649158 outside = 79.13142143513082 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 8:45:00 PM temperature> inside = 75.32714887649158 outside = 78.33355349529404 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:00:00 PM temperature> inside = 75.42714887649157 outside = 77.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:15:00 PM temperature> inside = 75.52714887649157 outside = 76.63433035328502 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:30:00 PM temperature> inside = 75.62714887649156 outside = 75.74025148547635 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 9:45:00 PM temperature> inside = 75.52714887649157 outside = 74.82159197954742 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:00:00 PM temperature> inside = 75.42714887649157 outside = 73.88228567653782 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:15:00 PM temperature> inside = 75.32714887649158 outside = 72.92635483024192 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:30:00 PM temperature> inside = 75.22714887649158 outside = 71.95789288330077 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 10:45:00 PM temperature> inside = 75.12714887649159 outside = 70.98104693845215 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:00:00 PM temperature> inside = 75.0271488764916 outside = 70 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:15:00 PM temperature> inside = 74.9271488764916 outside = 69.01895306154786 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:30:00 PM temperature> inside = 74.8271488764916 outside = 68.04210711669923 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/6/2021 11:45:00 PM temperature> inside = 74.72714887649161 outside = 67.07364516975808 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:00:00 AM temperature> inside = 74.62714887649162 outside = 66.11771432346218 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:15:00 AM temperature> inside = 74.52714887649162 outside = 65.17840802045258 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:30:00 AM temperature> inside = 74.42714887649163 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:45:00 AM temperature> inside = 74.32714887649163 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:00:00 AM temperature> inside = 74.22714887649164 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:15:00 AM temperature> inside = 74.12714887649165 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:30:00 AM temperature> inside = 74.02714887649165 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:45:00 AM temperature> inside = 73.92714887649166 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:00:00 AM temperature> inside = 73.82714887649166 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:15:00 AM temperature> inside = 73.72714887649167 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:30:00 AM temperature> inside = 73.62714887649167 outside = 58.09969989563147 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:45:00 AM temperature> inside = 73.52714887649168 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:00:00 AM temperature> inside = 73.42714887649169 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:15:00 AM temperature> inside = 73.32714887649169 outside = 56.54690887700967 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:30:00 AM temperature> inside = 73.2271488764917 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:45:00 AM temperature> inside = 73.1271488764917 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:00:00 AM temperature> inside = 73.02714887649171 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:15:00 AM temperature> inside = 72.92714887649171 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:30:00 AM temperature> inside = 72.82714887649172 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:45:00 AM temperature> inside = 72.72714887649173 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:00:00 AM temperature> inside = 72.62714887649173 outside = 55 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:15:00 AM temperature> inside = 72.52714887649174 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Engine failing over... Note we'll continue to see the producer's `temperature>` traces, but no `thermostat>` outputs.\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:30:00 AM temperature> inside = 72.42714887649174 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:45:00 AM temperature> inside = 72.32714887649175 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:00:00 AM temperature> inside = 72.22714887649175 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:15:00 AM temperature> inside = 72.12714887649176 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:30:00 AM temperature> inside = 72.02714887649176 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:45:00 AM temperature> inside = 71.92714887649177 outside = 56.54690887700968 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:00:00 AM temperature> inside = 71.82714887649178 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:15:00 AM temperature> inside = 71.72714887649178 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:30:00 AM temperature> inside = 71.62714887649179 outside = 58.099699895631474 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:45:00 AM temperature> inside = 71.5271488764918 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:00:00 AM temperature> inside = 71.4271488764918 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:15:00 AM temperature> inside = 71.3271488764918 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:30:00 AM temperature> inside = 71.22714887649181 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:45:00 AM temperature> inside = 71.12714887649182 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:00:00 AM temperature> inside = 71.02714887649182 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:15:00 AM temperature> inside = 70.92714887649183 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:30:00 AM temperature> inside = 70.82714887649183 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:45:00 AM temperature> inside = 70.72714887649184 outside = 65.17840802045258 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:00:00 AM temperature> inside = 70.62714887649184 outside = 66.1177143234622 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:15:00 AM temperature> inside = 70.52714887649185 outside = 67.07364516975808 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Engine recovered!\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:30:00 AM temperature> inside = 70.42714887649186 outside = 68.04210711669923 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:45:00 AM temperature> inside = 70.32714887649186 outside = 69.01895306154785 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:00:00 AM temperature> inside = 70.22714887649187 outside = 70 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:15:00 AM temperature> inside = 70.32714887649186 outside = 70.98104693845215 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:30:00 AM temperature> inside = 70.42714887649186 outside = 71.95789288330077 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:45:00 AM temperature> inside = 70.52714887649185 outside = 72.92635483024192 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:00:00 PM temperature> inside = 70.62714887649184 outside = 73.8822856765378 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:15:00 PM temperature> inside = 70.72714887649184 outside = 74.82159197954742 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:30:00 PM temperature> inside = 70.82714887649183 outside = 75.74025148547635 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 12:45:00 PM temperature> inside = 70.92714887649183 outside = 76.63433035328502 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:00:00 PM temperature> inside = 71.02714887649182 outside = 77.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:15:00 PM temperature> inside = 71.12714887649182 outside = 78.33355349529404 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:30:00 PM temperature> inside = 71.22714887649181 outside = 79.1314214351308 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 1:45:00 PM temperature> inside = 71.3271488764918 outside = 79.89018722650103 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:00:00 PM temperature> inside = 71.4271488764918 outside = 80.60660171779821 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:15:00 PM temperature> inside = 71.5271488764918 outside = 81.27759711218467 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:30:00 PM temperature> inside = 71.62714887649179 outside = 81.90030010436853 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 2:45:00 PM temperature> inside = 71.72714887649178 outside = 82.47204418453818 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:00:00 PM temperature> inside = 71.82714887649178 outside = 82.99038105676658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:15:00 PM temperature> inside = 71.92714887649177 outside = 83.45309112299033 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:30:00 PM temperature> inside = 72.02714887649176 outside = 83.8581929876693 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 3:45:00 PM temperature> inside = 72.12714887649176 outside = 84.20395194242658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:00:00 PM temperature> inside = 72.22714887649175 outside = 84.48888739433602 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:15:00 PM temperature> inside = 72.32714887649175 outside = 84.71177920604846 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:30:00 PM temperature> inside = 72.42714887649174 outside = 84.87167292060715 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 4:45:00 PM temperature> inside = 72.52714887649174 outside = 84.96788384857905 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:00:00 PM temperature> inside = 72.62714887649173 outside = 85 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:15:00 PM temperature> inside = 72.72714887649173 outside = 84.96788384857905 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:30:00 PM temperature> inside = 72.82714887649172 outside = 84.87167292060715 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 5:45:00 PM temperature> inside = 72.92714887649171 outside = 84.71177920604846 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:00:00 PM temperature> inside = 73.02714887649171 outside = 84.48888739433602 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:15:00 PM temperature> inside = 73.1271488764917 outside = 84.20395194242658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:30:00 PM temperature> inside = 73.2271488764917 outside = 83.8581929876693 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 6:45:00 PM temperature> inside = 73.32714887649169 outside = 83.45309112299033 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:00:00 PM temperature> inside = 73.42714887649169 outside = 82.99038105676658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:15:00 PM temperature> inside = 73.52714887649168 outside = 82.47204418453818 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:30:00 PM temperature> inside = 73.62714887649167 outside = 81.90030010436853 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 7:45:00 PM temperature> inside = 73.72714887649167 outside = 81.27759711218467 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:00:00 PM temperature> inside = 73.82714887649166 outside = 80.60660171779821 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:15:00 PM temperature> inside = 73.92714887649166 outside = 79.89018722650104 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:30:00 PM temperature> inside = 74.02714887649165 outside = 79.13142143513082 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 8:45:00 PM temperature> inside = 74.12714887649165 outside = 78.33355349529404 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:00:00 PM temperature> inside = 74.22714887649164 outside = 77.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:15:00 PM temperature> inside = 74.32714887649163 outside = 76.63433035328502 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:30:00 PM temperature> inside = 74.42714887649163 outside = 75.74025148547635 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 9:45:00 PM temperature> inside = 74.52714887649162 outside = 74.82159197954742 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:00:00 PM temperature> inside = 74.42714887649163 outside = 73.88228567653782 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:15:00 PM temperature> inside = 74.32714887649163 outside = 72.92635483024192 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:30:00 PM temperature> inside = 74.22714887649164 outside = 71.95789288330077 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 10:45:00 PM temperature> inside = 74.12714887649165 outside = 70.98104693845215 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:00:00 PM temperature> inside = 74.02714887649165 outside = 70 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:15:00 PM temperature> inside = 73.92714887649166 outside = 69.01895306154786 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:30:00 PM temperature> inside = 73.82714887649166 outside = 68.04210711669923 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/7/2021 11:45:00 PM temperature> inside = 73.72714887649167 outside = 67.07364516975808 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:00:00 AM temperature> inside = 73.62714887649167 outside = 66.11771432346218 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:15:00 AM temperature> inside = 73.52714887649168 outside = 65.17840802045258 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:30:00 AM temperature> inside = 73.42714887649169 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:45:00 AM temperature> inside = 73.32714887649169 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:00:00 AM temperature> inside = 73.2271488764917 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:15:00 AM temperature> inside = 73.1271488764917 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:30:00 AM temperature> inside = 73.02714887649171 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:45:00 AM temperature> inside = 72.92714887649171 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:00:00 AM temperature> inside = 72.82714887649172 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:15:00 AM temperature> inside = 72.72714887649173 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:30:00 AM temperature> inside = 72.62714887649173 outside = 58.09969989563147 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:45:00 AM temperature> inside = 72.52714887649174 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:00:00 AM temperature> inside = 72.42714887649174 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:15:00 AM temperature> inside = 72.32714887649175 outside = 56.54690887700967 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:30:00 AM temperature> inside = 72.22714887649175 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:45:00 AM temperature> inside = 72.12714887649176 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 4:00:00 AM temperature> inside = 72.02714887649176 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 4:15:00 AM temperature> inside = 71.92714887649177 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 4:30:00 AM temperature> inside = 71.82714887649178 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 4:45:00 AM temperature> inside = 71.72714887649178 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 5:00:00 AM temperature> inside = 71.62714887649179 outside = 55 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 5:15:00 AM temperature> inside = 71.5271488764918 outside = 55.03211615142095 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 5:30:00 AM temperature> inside = 71.4271488764918 outside = 55.12832707939285 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 5:45:00 AM temperature> inside = 71.3271488764918 outside = 55.288220793951545 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 6:00:00 AM temperature> inside = 71.22714887649181 outside = 55.511112605663975 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 6:15:00 AM temperature> inside = 71.12714887649182 outside = 55.796048057573415 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 6:30:00 AM temperature> inside = 71.02714887649182 outside = 56.1418070123307 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 6:45:00 AM temperature> inside = 70.92714887649183 outside = 56.54690887700968 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 7:00:00 AM temperature> inside = 70.82714887649183 outside = 57.00961894323342 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 7:15:00 AM temperature> inside = 70.72714887649184 outside = 57.52795581546182 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 7:30:00 AM temperature> inside = 70.62714887649184 outside = 58.099699895631474 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 7:45:00 AM temperature> inside = 70.52714887649185 outside = 58.722402887815335 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 8:00:00 AM temperature> inside = 70.42714887649186 outside = 59.39339828220179 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 8:15:00 AM temperature> inside = 70.32714887649186 outside = 60.109812773498966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 8:30:00 AM temperature> inside = 70.22714887649187 outside = 60.86857856486919 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 8:45:00 AM temperature> inside = 70.12714887649187 outside = 61.666446504705966 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 9:00:00 AM temperature> inside = 70.02714887649188 outside = 62.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 9:15:00 AM temperature> inside = 69.92714887649188 outside = 63.36566964671498 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 9:30:00 AM temperature> inside = 69.82714887649189 outside = 64.25974851452365 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 9:30:00 AM thermostat> ON heating to 75\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 9:45:00 AM temperature> inside = 69.92714887649188 outside = 65.17840802045258 target = 75\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 10:00:00 AM temperature> inside = 70.02714887649188 outside = 66.1177143234622 target = 75\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 10:15:00 AM temperature> inside = 70.12714887649187 outside = 67.07364516975808 target = 75\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 10:30:00 AM temperature> inside = 70.22714887649187 outside = 68.04210711669923 target = 75\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 10:30:00 AM thermostat> OFF\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 10:45:00 AM temperature> inside = 70.12714887649187 outside = 69.01895306154785 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 11:00:00 AM temperature> inside = 70.02714887649188 outside = 70 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 11:15:00 AM temperature> inside = 70.12714887649187 outside = 70.98104693845215 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 11:30:00 AM temperature> inside = 70.22714887649187 outside = 71.95789288330077 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 11:45:00 AM temperature> inside = 70.32714887649186 outside = 72.92635483024192 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:00:00 PM temperature> inside = 70.42714887649186 outside = 73.8822856765378 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:15:00 PM temperature> inside = 70.52714887649185 outside = 74.82159197954742 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:30:00 PM temperature> inside = 70.62714887649184 outside = 75.74025148547635 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 12:45:00 PM temperature> inside = 70.72714887649184 outside = 76.63433035328502 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:00:00 PM temperature> inside = 70.82714887649183 outside = 77.5 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:15:00 PM temperature> inside = 70.92714887649183 outside = 78.33355349529404 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:30:00 PM temperature> inside = 71.02714887649182 outside = 79.1314214351308 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 1:45:00 PM temperature> inside = 71.12714887649182 outside = 79.89018722650103 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:00:00 PM temperature> inside = 71.22714887649181 outside = 80.60660171779821 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:15:00 PM temperature> inside = 71.3271488764918 outside = 81.27759711218467 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:30:00 PM temperature> inside = 71.4271488764918 outside = 81.90030010436853 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 2:45:00 PM temperature> inside = 71.5271488764918 outside = 82.47204418453818 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:00:00 PM temperature> inside = 71.62714887649179 outside = 82.99038105676658 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:15:00 PM temperature> inside = 71.72714887649178 outside = 83.45309112299033 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "3/8/2021 3:30:00 PM temperature> inside = 71.82714887649178 outside = 83.8581929876693 target = \r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Disposing subscription...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Stopping simulator...\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        },
        {
          "data": {
            "text/plain": "Done!\r\n"
          },
          "execution_count": 1,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "var stopEventProducer = new CancellationTokenSource();\n",
        "\n",
        "Console.WriteLine(\"Starting simulator for temperature sensor readings...\");\n",
        "\n",
        "var logger = SubscribeToSettingsStream();\n",
        "var producer = RunReadingsGenerator(stopEventProducer.Token);\n",
        "\n",
        "var subUri = new Uri(\"iot://reactor/subscription/BD/livingroom/comfy\");\n",
        "\n",
        "Console.WriteLine(\"Setting up query engine...\");\n",
        "\n",
        "await WithEngine(async engine =>\n",
        "{\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    var input = ctx.GetObservable<string, SensorReading>(new Uri(\"iot://reactor/observables/ingress\"));\n",
        "    var output = ctx.GetObserver<string, double?>(new Uri(\"iot://reactor/observers/egress\"));\n",
        "\n",
        "    var readings = input(\"bart://sensors/home/livingroom/temperature/readings\");\n",
        "    var settings = output(\"bart://sensors/home/livingroom/temperature/settings\");\n",
        "\n",
        "    Console.WriteLine(\"Creating subscription...\");\n",
        "\n",
        "    await readings.Window(4).SelectMany(w => w.Average(r => r.Temperature)).Select(t => t < 70 || t > 80 ? 75 : default(double?)).DistinctUntilChanged().SubscribeAsync(settings, subUri, null, CancellationToken.None);\n",
        "\n",
        "    // Run for a bit.\n",
        "    await Task.Delay(TimeSpan.FromSeconds(30));\n",
        "\n",
        "    Console.WriteLine(\"Engine failing over... Note we'll continue to see the producer's `temperature>` traces, but no `thermostat>` outputs.\");\n",
        "});\n",
        "\n",
        "await Task.Delay(TimeSpan.FromSeconds(5));\n",
        "\n",
        "await WithEngine(async engine =>\n",
        "{\n",
        "    Console.WriteLine(\"Engine recovered!\");\n",
        "\n",
        "    var ctx = new ReactorContext(engine);\n",
        "\n",
        "    // Run for a bit more.\n",
        "    await Task.Delay(TimeSpan.FromSeconds(30));\n",
        "\n",
        "    Console.WriteLine(\"Disposing subscription...\");\n",
        "\n",
        "    await ctx.GetSubscription(subUri).DisposeAsync();\n",
        "});\n",
        "\n",
        "Console.WriteLine(\"Stopping simulator...\");\n",
        "\n",
        "stopEventProducer.Cancel();\n",
        "producer.Wait();\n",
        "\n",
        "logger.Dispose();\n",
        "\n",
        "Console.WriteLine(\"Done!\");"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": ".NET (C#)",
      "language": "C#",
      "name": ".net-csharp"
    },
    "language_info": {
      "file_extension": ".cs",
      "mimetype": "text/x-csharp",
      "name": "C#",
      "pygments_lexer": "csharp",
      "version": "9.0"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 4
}